package com.crazyice.lee.reader;

import com.crazyice.lee.data.Student;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;

import java.io.IOException;
import java.io.InputStream;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.util.Properties;

/**
 * @Description mysql source
 * @Author jiangxiaozhi
 * @Date 2018/10/15 17:05
 **/
@Slf4j
public class JdbcReader extends RichSourceFunction<Student> {
    private Connection connection = null;
    private PreparedStatement ps = null;

    private static Properties properties;

    private static void init() {
        properties = new Properties();
        try(InputStream inputStream = JdbcReader.class.getResourceAsStream("/jdbc.properties")) {
            properties.load(inputStream);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    //该方法主要用于打开数据库连接，下面的ConfigKeys类是获取配置的类
    @Override
    public void open(Configuration parameters) throws Exception {
        init();

        super.open(parameters);
        Class.forName(properties.getProperty("DriverClass"));
        connection = DriverManager.getConnection(properties.getProperty("SourceDriverUrl"), properties.getProperty("SourceUser"), properties.getProperty("SourcePassword"));
        ps = connection.prepareStatement(properties.getProperty("SourceSql"));
        super.open(parameters);
    }
 
    //执行查询并获取结果
    @Override
    public void run(SourceContext<Student> ctx) throws Exception {
        try {
            ResultSet resultSet = ps.executeQuery();
            while (resultSet.next()) {
                int id = resultSet.getInt("id");
                String name = resultSet.getString("name");
                int age=resultSet.getInt("age");
                String password=resultSet.getString("password");
                boolean sex=resultSet.getBoolean("sex");
                Student student=new Student(id,name,password,sex,age);
                ctx.collect(student);
            }
        } catch (Exception e) {
            log.error("runException:{}", e);
        }
 
    }
     
    //关闭数据库连接
    @Override
    public void cancel() {
        try {
            super.close();
            if (connection != null) {
                connection.close();
            }
            if (ps != null) {
                ps.close();
            }
        } catch (Exception e) {
            log.error("runException:{}", e);
        }
    }
}